package console

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"

	"github.com/louloulin/dataflare/pkg/core"
	"github.com/louloulin/dataflare/pkg/messages"
)

// ConsoleSinkConnector 是一个控制台接收器连接器
type ConsoleSinkConnector struct {
	// format 是输出格式
	format string
	// mu 是保护连接器的互斥锁
	mu sync.RWMutex
	// running 表示连接器是否正在运行
	running bool
}

// NewConsoleSinkConnector 创建一个新的控制台接收器连接器
func NewConsoleSinkConnector() (core.SinkConnector, error) {
	return &ConsoleSinkConnector{}, nil
}

// Init 初始化连接器
func (c *ConsoleSinkConnector) Init(ctx context.Context, config map[string]interface{}) error {
	// 从配置中获取格式
	formatRaw, ok := config["format"]
	if !ok {
		c.format = "json"
	} else {
		format, ok := formatRaw.(string)
		if !ok {
			return fmt.Errorf("format must be a string")
		}
		c.format = format
	}

	// 验证格式
	if c.format != "json" && c.format != "text" {
		return fmt.Errorf("invalid format: %s", c.format)
	}

	return nil
}

// Start 启动连接器
func (c *ConsoleSinkConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.running = true
	return nil
}

// Stop 停止连接器
func (c *ConsoleSinkConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.running = false
	return nil
}

// Write 向接收器写入记录
func (c *ConsoleSinkConnector) Write(ctx context.Context, record messages.Record) error {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// 检查是否正在运行
	if !c.running {
		return fmt.Errorf("connector not running")
	}

	// 格式化输出
	var output string
	switch c.format {
	case "json":
		// 解析JSON
		var data interface{}
		err := json.Unmarshal(record.Value, &data)
		if err != nil {
			return fmt.Errorf("failed to parse JSON: %w", err)
		}

		// 格式化JSON
		formatted, err := json.MarshalIndent(data, "", "  ")
		if err != nil {
			return fmt.Errorf("failed to format JSON: %w", err)
		}

		output = string(formatted)
	case "text":
		output = string(record.Value)
	}

	// 打印到控制台
	fmt.Println(output)
	return nil
}

// Flush 刷新缓冲记录
func (c *ConsoleSinkConnector) Flush(ctx context.Context) error {
	// 没有缓冲，所以不需要刷新
	return nil
}

// 注册连接器
func init() {
	// 注册控制台接收器连接器
	core.RegisterSinkConnector("console", NewConsoleSinkConnector)
}
